#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Licensed to Cloudera, Inc. under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  Cloudera, Inc. licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from future import standard_library
standard_library.install_aliases()
from builtins import zip, range, object

from django.utils.encoding import smart_str

import json
import logging
import os
import pytest
import random
import re
import tempfile
import threading

from desktop.redaction.engine import RedactionEngine, RedactionPolicy, RedactionRule, parse_redaction_policy_from_file, \
    _convert_java_pattern_to_python
from desktop.redaction.logfilter import add_log_redaction_filter_to_logger


MESSAGE = "This string is not redacted"


def get_path(filename):
  return os.path.join(os.path.dirname(__file__), 'test_data', filename)


class MockLoggingHandler(logging.Handler):
  def __init__(self, *args, **kwargs):
    logging.Handler.__init__(self, *args, **kwargs)

    self.records = []

  def emit(self, record):
    self.records.append(record)

  def reset(self):
    del self.records[:]


class TestRedactionRule(object):
  def test_redaction_rule_works(self):
    rule = RedactionRule('password=', 'password=".*"', 'password="???"')

    test_strings = [
        ('message', 'message'),
        ('password="a password"', 'password="???"'),
        ('before password="a password" after', 'before password="???" after'),
    ]

    for message, redacted_message in test_strings:
      assert rule.redact(message) == redacted_message

  def test_non_redacted_string_returns_same_string(self):
    rule = RedactionRule('password=', 'password=".*"', 'password="???"')

    message = 'message'
    assert rule.redact(message) is message

  def test_equality(self):
    rule1 = RedactionRule('password=', 'password=".*"', 'password="???"')
    rule2 = RedactionRule('password=', 'password=".*"', 'password="???"')
    rule3 = RedactionRule('ssn=', 'ssn=\d{3}-\d{2}-\d{4}', 'ssn=XXX-XX-XXXX'),

    assert rule1 == rule2
    assert rule1 != rule3


  def test_parse_redaction_policy_from_file(self):
    with tempfile.NamedTemporaryFile(mode='w') as f:
      json.dump({
          'version': 1,
          'rules': [
            {
              'description': 'redact passwords',
              'trigger': 'password=',
              'search': 'password=".*"',
              'replace': 'password="???"',
            },
            {
              'description': 'redact social security numbers',
              'search': '\d{3}-\d{2}-\d{4}',
              'replace': 'XXX-XX-XXXX',
            },
          ]
      }, f)

      f.flush()

      policy = parse_redaction_policy_from_file(f.name)

      assert policy.rules == [
        RedactionRule(u'password=', u'password=".*"', u'password="???"'),
        RedactionRule(None, u'\d{3}-\d{2}-\d{4}', u'XXX-XX-XXXX'),
      ]


class TestRedactionEngine(object):
  def test_redaction_works(self):
    redaction_engine = RedactionEngine([
      RedactionRule('password=', 'password=".*"', 'password="???"'),
      RedactionRule('ssn=', 'ssn=\d{3}-\d{2}-\d{4}', 'ssn=XXX-XX-XXXX'),
    ])

    test_strings = [
        ('message', 'message'),
        ('password="a password"', 'password="???"'),
        ('before password="a password" after', 'before password="???" after'),
        ('an ssn=123-45-6789', 'an ssn=XXX-XX-XXXX'),
    ]

    for message, redacted_message in test_strings:
      assert redaction_engine.redact(message) == redacted_message

  def test_equality(self):
    engine1 = RedactionEngine([
        RedactionRule('password=', 'password=".*"', 'password="???"'),
    ])
    engine2 = RedactionEngine([
        RedactionRule('password=', 'password=".*"', 'password="???"'),
    ])
    engine3 = RedactionEngine([
        RedactionRule('ssn=', 'ssn=\d{3}-\d{2}-\d{4}', 'ssn=XXX-XX-XXXX'),
    ])

    assert engine1 == engine2
    assert engine1 != engine3

  def test_redact_empty_string(self):
    engine = RedactionEngine([
        RedactionRule('password=', 'password=".*"', 'password="???"'),
    ])

    assert engine.redact(None) == None
    assert engine.redact('') == ''


class TestRedactionLogFilter(object):

  @classmethod
  def setup_class(cls):
    cls.logger = logging.getLogger(cls.__name__)

    cls.handler = MockLoggingHandler()
    cls.logger.addHandler(cls.handler)

    policy = RedactionPolicy([
      RedactionRule('password=', 'password=".*"', 'password="???"'),
      RedactionRule('ssn=', 'ssn=\d{3}-\d{2}-\d{4}', 'ssn=XXX-XX-XXXX'),
    ])

    engine = RedactionEngine([policy])

    add_log_redaction_filter_to_logger(engine, cls.logger)

  @classmethod
  def teardown_class(cls):
    cls.logger.handlers = []

  def teardown_method(self):
    self.handler.reset()

  def test_redaction_filter(self):
    test_strings = [
        {
          'message': 'message',
          'result_message': 'message',
          'result_msg': 'message',
          'result_args': (),
        },
        {
          'message': 'message %s',
          'args': ['an arg'],
          'result_message': 'message an arg',
          'result_msg': 'message %s',
          'result_args': ('an arg',),
        },
        {
          'message': 'password="a password"',
          'result_message': 'password="???"',
        },
        {
          'message': 'password="%s"',
          'args': ['a password'],
          'result_message': 'password="???"',
        },
        {
          'message': 'password=%s',
          'args': ['"a password"'],
          'result_message': 'password="???"',
        },
        {
          'message': 'before password="%s" after',
          'args': ['a password'],
          'result_message': 'before password="???" after',
        },

        {
          'message': 'ssn=%s-%s-%s',
          'args': ['123', '45', '6789'],
          'result_message': 'ssn=XXX-XX-XXXX',
        },
    ]

    for test in test_strings:
      self.logger.debug(test['message'], *test.get('args', ()))

    for test, record in zip(test_strings, self.handler.records):
      assert record.getMessage() == test['result_message']
      assert record.message == test['result_message']
      assert record.msg == test.get('result_msg', test['result_message'])
      assert record.args == test.get('result_args')

  def test_convert_java_pattern_to_python(self):
    assert _convert_java_pattern_to_python('1-2') == '1-2'
    assert _convert_java_pattern_to_python('$1-$2') == '\\1-\\2'
    assert _convert_java_pattern_to_python('\\$1-$2') == '$1-\\2'
    assert _convert_java_pattern_to_python('\\$$1-$2') == '$\\1-\\2'

  def test_does_not_exist(self):
    path = get_path('thisfiledoesnotexist.json')
    with pytest.raises(IOError):
      parse_redaction_policy_from_file(path)

  def test_is_dir(self):
    path = '/tmp'
    with pytest.raises(IOError):
      parse_redaction_policy_from_file(path)

  def test_is_not_json(self):
    path = get_path('not-json.json')
    with pytest.raises(IOError):
      parse_redaction_policy_from_file(path)

  def test_no_version(self):
    path = get_path('no-version.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_unknown_version(self):
    path = get_path('unknown-version.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_alpha_version(self):
    path = get_path('alpha-version.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_no_search(self):
    path = get_path('no-search.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_no_replace(self):
    path = get_path('no-replace.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_no_brace(self):
    path = get_path('no-brace.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_bad_regex(self):
    path = get_path('bad-regex.json')
    with pytest.raises(re.error):
      parse_redaction_policy_from_file(path)

  def test_extra_attr(self):
    path = get_path('extra-attr.json')
    with pytest.raises(ValueError):
      parse_redaction_policy_from_file(path)

  def test_empty_file(self):
    path = get_path('empty.json')
    policy = parse_redaction_policy_from_file(path)
    assert MESSAGE == policy.redact(MESSAGE)

  def test_empty_rules(self):
    path = get_path('empty-rules.json')
    policy = parse_redaction_policy_from_file(path)
    assert MESSAGE == policy.redact(MESSAGE)

  def test_basic_good1(self):
    path = get_path('good-1.json')
    policy = parse_redaction_policy_from_file(path)
    assert "Hxllx, wxrld" == policy.redact("Hello, world")

  def test_int_version(self):
    path = get_path('verint.json')
    policy = parse_redaction_policy_from_file(path)
    assert "Hxllx, wxrld" == policy.redact("Hello, world")

  def test_real_rules(self):
    path = get_path('real-1.json')
    policy = parse_redaction_policy_from_file(path)

    messages = [
      ("Hello, world", "Hello, world"),
      ("CC 1234-2345-3456-4576", "CC XXXX-XXXX-XXXX-XXXX"),
      ("CC 1234234534654576", "CC XXXXXXXXXXXXXXXX"),
      ("CC 1234,2345,3456,4576", "CC XXXX-XXXX-XXXX-XXXX"),
      ("SSN 123-45-6789", "SSN XXX-XX-XXXX"),
      ("SSN 123456789", "SSN XXXXXXXXX"),
      ("My password=Hello123", "My password=xxxxx"),
      ("Host www.cloudera.com", "Host HOSTNAME.REDACTED"),
      ("www.c1-foo.org rules!", "HOSTNAME.REDACTED rules!"),
      ("IP1 8.8.8.8", "IP1 0.0.0.0"),
      ("IP2 192.168.0.1", "IP2 0.0.0.0"),
      ("My email is myoder@cloudera.com", "My email is email@redacted.host"),
      ("hello.world@ex.x-1.fr is interesting", "email@redacted.host is interesting"),
      ("Multi 1234-2345-3456-4567\nLine 123-45-6789", "Multi XXXX-XXXX-XXXX-XXXX\nLine XXX-XX-XXXX"),
    ]

    for message, redacted_message in messages:
      assert redacted_message == policy.redact(message)

  def test_unicode_strings(self):
    path = get_path('real-1.json')
    policy = parse_redaction_policy_from_file(path)

    messages = [
      ("äöüß 123-45-6789", "äöüß XXX-XX-XXXX"),
      ("你好阿基尔 1234234534654576", "你好阿基尔 XXXXXXXXXXXXXXXX"),
      ("ã 你好 1234,2345,3456,4576", "ã 你好 XXXX-XXXX-XXXX-XXXX"),
    ]

    for message, redacted_message in messages:
      message_to_redact = smart_str(message)
      self.logger.debug("Message to redact : %s " % message_to_redact)
      self.logger.debug("Message after redact : %s " % policy.redact(message_to_redact))
      assert redacted_message == policy.redact(message_to_redact)

  def test_huge_rules(self):
    path = get_path('huge-1.json')
    policy = parse_redaction_policy_from_file(path)
    assert "This string is not redadted" == policy.redact(MESSAGE)

  def test_back_refs(self):
    path = get_path('replace-1.json')
    policy = parse_redaction_policy_from_file(path)

    messages = [
      ("Hello, world", "Hello, world"),
      ("1234-2345-3456-4576", "XXXX-XXXX-XXXX-4576"),
      ("Words www.gmail.com is cool", "Words HOSTNAME.REDACTED.com is cool"),
      ("short.org", "HOSTNAME.REDACTED.org"),
      ("long.n4me.h-1.co.fr", "HOSTNAME.REDACTED.fr"),
      ("Ping 192.168.0.1", "Ping 0.192.1.168"),
      ("Magic word", "word: Magic word, word"),
    ]

    for message, redacted_message in messages:
      assert redacted_message == policy.redact(message)

  def test_ordering(self):
    path = get_path('ordering-1.json')
    policy = parse_redaction_policy_from_file(path)

    messages = [
      ("Hello, world", "Hello, world"),
      ("one", "four"),
      ("This one is a nice one", "This four is a nice four"),
      ("Please help me: ten", "Please help me: thirteen"),
      ("HappY abc", "HappY stu"),
    ]

    for message, redacted_message in messages:
      assert redacted_message == policy.redact(message)

  def test_case_sensitivity(self):
    path = get_path('case-1.json')
    policy = parse_redaction_policy_from_file(path)

    messages = [
      ("Hello, world", "Hello, world"),
      ("Say aAa! aaa! AAAAAA!", "Say bbb! bbb! bbbbbb!"),
      ("I like dddogs. dDd", "I like dddogs. dDd"),
      ("Cccats. Dddogs", "Cccats. eEeogs"),
      ("Trigger fff gGg", "Trigger fff gGg"),
      ("Trigger fFf Ggg", "Trigger fFf Ggg"),
      ("Trigger fFf gGg", "Trigger fFf hHh"),
    ]

    for message, redacted_message in messages:
      assert redacted_message == policy.redact(message)

  def test_multithreading(self):
    path = get_path('numbers.json')
    policy = parse_redaction_policy_from_file(path)

    assert "asdf####fdas### H#ll# w#rld" == policy.redact("asdf1234fdas666 H3ll0 w0rld")

    errors = []
    lock = threading.Lock()

    regex = re.compile(r"[0-9]")

    class TestThread(threading.Thread):
      def run(self):
        for i in range(500):
          message = u''.join(random_utf8_char() for _ in range(128))
          redacted_message = policy.redact(message)

          if regex.search(redacted_message):
            with lock:
              errors.append((message, redacted_message))
              break

    threads = []
    for i in range(10):
      threads.append(TestThread())

    for thread in threads:
      thread.start()

    for thread in threads:
      thread.join()

    assert errors == []

def byte_range(first, last):
  return list(range(first, last+1))

first_values = byte_range(0x00, 0x7F) + byte_range(0xC2, 0xF4)
trailing_values = byte_range(0x80, 0xBF)

def random_utf8_char():
  first = random.choice(first_values)
  if first <= 0x7F:
    value = bytearray([first])
  elif first <= 0xDF:
    value = bytearray([first, random.choice(trailing_values)])
  elif first == 0xE0:
    value = bytearray([first, random.choice(byte_range(0xA0, 0xBF)), random.choice(trailing_values)])
  elif first == 0xED:
    value = bytearray([first, random.choice(byte_range(0x80, 0x9F)), random.choice(trailing_values)])
  elif first <= 0xEF:
    value = bytearray([first, random.choice(trailing_values), random.choice(trailing_values)])
  elif first == 0xF0:
    value = bytearray([first, random.choice(byte_range(0x90, 0xBF)), random.choice(trailing_values), random.choice(trailing_values)])
  elif first <= 0xF3:
    value = bytearray([first, random.choice(trailing_values), random.choice(trailing_values), random.choice(trailing_values)])
  elif first == 0xF4:
    value = bytearray([first, random.choice(byte_range(0x80, 0x8F)), random.choice(trailing_values), random.choice(trailing_values)])
  return value.decode('utf8')
